package net.bwie.realtime.gongdan.dws.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * @Author: FuHe
 * @Date: 2025/6/10
 * todo 大数据-用户画像-12-达摩盘家庭特征
 */
public class GongDanLabelGenderJob {
    public static void main(String[] args) {
        // 表执行环境
        TableEnvironment tabEnv = getTableEnv();
        // 输入表  input 映射到kafka消息队列
        createInputTable(tabEnv);
        // 数据处理
        Table reportTable = handle(tabEnv);
        reportTable.execute().print();
        // 输出表  output  映射到clickhouse表
//        createOutputTable(tabEnv);
        // 保存数据
//        saveToDoris(tabEnv, reportTable);
    }

//    private static void saveToDoris(TableEnvironment tabEnv, Table reportTable) {
//
//    }
//
    private static void createOutputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("");
    }

    private static Table handle(TableEnvironment tabEnv) {
        Table table = tabEnv.sqlQuery("SELECT\n" +
                "        user_id,\n" +
                "        item_id,\n" +
                "        CAST(CASE\n" +
                "            WHEN (UNIX_TIMESTAMP(DATE_FORMAT(CURRENT_TIMESTAMP, 'yyyy-MM-dd HH:mm:ss'), 'yyyy-MM-dd HH:mm:ss') -\n" +
                "                  UNIX_TIMESTAMP(behavior_time, 'yyyy-MM-dd HH:mm:ss'))/60/60/24 <= 30 THEN 1.0\n" +
                "            WHEN (UNIX_TIMESTAMP(DATE_FORMAT(CURRENT_TIMESTAMP, 'yyyy-MM-dd HH:mm:ss'), 'yyyy-MM-dd HH:mm:ss') -\n" +
                "                  UNIX_TIMESTAMP(behavior_time, 'yyyy-MM-dd HH:mm:ss'))/60/60/24 <= 60 THEN 0.5\n" +
                "            ELSE 0.2\n" +
                "            END AS DECIMAL(10, 2)) AS time_decay,\n" +
                "        CAST(CASE behavior_type\n" +
                "            WHEN 'purchase' THEN 1.0\n" +
                "            WHEN 'search' THEN 0.8\n" +
                "            WHEN 'add_cart' THEN 0.4\n" +
                "            WHEN 'browse' THEN 0.2\n" +
                "            ELSE 0\n" +
                "            END AS DECIMAL(10, 2)) AS behavior_weight\n" +
                "    FROM user_behavior_clickhouse");
        tabEnv.createTemporaryView("table2", table);

        Table table2 = tabEnv.sqlQuery("SELECT\n" +
                "        item_id,\n" +
                "        CASE\n" +
                "            WHEN keywords LIKE '%女%' THEN 1.0\n" +
                "            WHEN keywords LIKE '%男%' THEN 1.0\n" +
                "            WHEN keywords LIKE '%中性%' THEN 0.5\n" +
                "            ELSE 0.2\n" +
                "        END AS category_weight\n" +
                "    FROM item_attributes_clickhouse");
        tabEnv.createTemporaryView("table3", table2);

        Table table3 = tabEnv.sqlQuery("SELECT\n" +
                "        user_id,\n" +
                "        COUNT(behavior_type) acount\n" +
                "    FROM user_behavior_clickhouse\n" +
                "    GROUP BY user_id");
        tabEnv.createTemporaryView("table4", table3);

        Table table4 = tabEnv.sqlQuery("SELECT\n" +
                "table2.user_id AS user_id," +
                "IF(sum(case when ui.gender='1' then table2.time_decay*table2.behavior_weight*table3.category_weight*table4.acount end) IS NULL, 0," +
                "   sum(case when ui.gender='1' then table2.time_decay*table2.behavior_weight*table3.category_weight*table4.acount end)) as maleScore,\n" +
                "IF(sum(case when ui.gender='2' then table2.time_decay*table2.behavior_weight*table3.category_weight*table4.acount end) IS NULL, 0," +
                "   sum(case when ui.gender='2' then table2.time_decay*table2.behavior_weight*table3.category_weight*table4.acount end)) as femaleScore\n" +
                "FROM table2\n" +
                "   left join table3  on table2.item_id = table3.item_id\n" +
                "   left join table4  on table2.user_id = table4.user_id\n" +
                "   left join user_info_clickhouse AS ui on table2.user_id = ui.user_id\n" +
                "group by  table2.user_id");
        tabEnv.createTemporaryView("table5", table4);

        Table reportTable = tabEnv.sqlQuery("SELECT\n" +
                "      user_id,\n" +
                "      CASE\n" +
                "          WHEN maleScore>femaleScore THEN '男'\n" +
                "          WHEN femaleScore>maleScore THEN '女'\n" +
                "          ELSE '未识别'\n" +
                "      end AS gender\n" +
                "FROM table5");
        return reportTable;
    }


    private static void createInputTable(TableEnvironment tabEnv) {
        // 商品属性表 - 添加更多配置参数
        tabEnv.executeSql("CREATE TABLE item_attributes_clickhouse (\n" +
                "    `item_id`       DECIMAL(20, 0) COMMENT '商品ID',\n" +
                "    `item_name`     STRING COMMENT '商品名称',\n" +
                "    `category_id`   STRING COMMENT '类目ID',\n" +
                "    `category_path` STRING COMMENT '完整类目路径',\n" +
                "    `keywords`      STRING COMMENT '商品关键词',\n" +
                "    `description`   STRING COMMENT '商品描述',\n" +
                "    `price`         DECIMAL(10, 2) COMMENT '商品价格',\n" +
                "    `is_maternal`   TINYINT COMMENT '是否母婴类商品',\n" +
                "    PRIMARY KEY (item_id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "    'connector' = 'clickhouse',\n" +
                "    'url' = 'jdbc:clickhouse://node103:8123/gongdan12',\n" +
                "    'table' = 'item_attributes',\n" +
                "    'username' = 'default',\n" +
                "    'password' = '',\n" +
                "    'format' = 'json'\n" +
                ")");

         // 用户表 - 添加更多配置参数
        tabEnv.executeSql("CREATE TABLE IF NOT EXISTS user_info_clickhouse (\n" +
                "    user_id       DECIMAL(20, 0) COMMENT '用户ID',\n" +
                "    register_time STRING COMMENT '注册时间',\n" +
                "    gender        TINYINT COMMENT '用户性别:1男/2女/0未知',\n" +
                "    age_range     STRING COMMENT '用户年龄段:18-24/25-30/...',\n" +
                "    province      STRING COMMENT '所在省份',\n" +
                "    city          STRING COMMENT '所在城市',\n" +
                "    is_parent     TINYINT COMMENT '是否家长身份'\n" +
                ") WITH (\n" +
                "    'connector' = 'clickhouse',\n" +
                "    'url' = 'jdbc:clickhouse://node103:8123/gongdan12',\n" +
                "    'table' = 'user_info',\n" +
                "    'username' = 'default',\n" +
                "    'password' = '',\n" +
                "    'format' = 'json'\n" +
                ")");

         // 订单明细表 - 添加更多配置参数
        tabEnv.executeSql("CREATE TABLE IF NOT EXISTS order_detail_clickhouse (\n" +
                "    order_id       DECIMAL(20, 0) COMMENT '订单ID',\n" +
                "    user_id        DECIMAL(20, 0) COMMENT '用户ID',\n" +
                "    item_id        DECIMAL(20, 0) COMMENT '商品ID',\n" +
                "    purchase_time  STRING COMMENT '购买时间',\n" +
                "    quantity       int COMMENT '购买数量',\n" +
                "    amount         DECIMAL(10, 2) COMMENT '订单金额',\n" +
                "    payment_method STRING COMMENT '支付方式',\n" +
                "    status         int COMMENT '订单状态:1已完成/2已退款'\n" +
                ") WITH (\n" +
                "    'connector' = 'clickhouse',\n" +
                "    'url' = 'jdbc:clickhouse://node103:8123/gongdan12',\n" +
                "    'table' = 'order_transaction',\n" +
                "    'username' = 'default',\n" +
                "    'password' = '',\n" +
                "    'format' = 'json'\n" +
                ")");

        // 用户行为表 - 添加更多配置参数
        tabEnv.executeSql("CREATE TABLE IF NOT EXISTS user_behavior_clickhouse (\n" +
                "    log_id         DECIMAL(20, 0) COMMENT '日志ID',\n" +
                "    user_id        DECIMAL(20, 0) COMMENT '用户ID',\n" +
                "    item_id        DECIMAL(20, 0) COMMENT '商品ID',\n" +
                "    behavior_type  STRING COMMENT '行为类型:purchase/browse/collect/add_cart/search',\n" +
                "    behavior_time  STRING COMMENT '行为时间',\n" +
                "    search_keyword STRING COMMENT '搜索关键词',\n" +
                "    session_id     STRING COMMENT '会话ID',\n" +
                "    stay_duration  bigint COMMENT '停留时长(ms)',\n" +
                "    page_url       STRING COMMENT '页面URL'\n" +
                ") WITH (\n" +
                "    'connector' = 'clickhouse',\n" +
                "    'url' = 'jdbc:clickhouse://node103:8123/gongdan12',\n" +
                "    'table' = 'user_behavior',\n" +
                "    'username' = 'default',\n" +
                "    'password' = '',\n" +
                "    'format' = 'json'\n" +
                ")");
    }

    private static TableEnvironment getTableEnv() {
        // 环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        // 时区设置
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        // 并发设置
        configuration.setString("table.exec.resource.default-parallelism", "1");
        // 状态保存时间
        configuration.setString("table.exec.state.ttl", "5s");
        // 返回值
        return tabEnv;
    }
}
